package com.ku.platformpay.service.mq;

import cn.hutool.json.JSONUtil;
import com.ku.entity.CustomerAccount;
import com.ku.entity.PlatformAccount;
import com.ku.platformpay.mapper.PlatformAccountMapper;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.io.UnsupportedEncodingException;
import java.math.BigDecimal;
import java.util.Date;
import java.util.List;
import java.util.Map;

/**
 * @program: rocketMQKill
 * @description:
 * @author: JokerKu
 * @create: 2020-05-07 10:01
 **/
@Component
public class TransactionMQConsumer {

    private DefaultMQPushConsumer consumer;

    private static final String CONSUMER_GROUP_NAME = "tx_pay_consumer_group";

    private static final String NAME_SERVER = "192.168.43.10:9876;192.168.43.9:9876";

    private static final String TX_PAY_TOPIC = "tx_pay_topic";

    public static final String TX_PAY_TAGS = "pay";

    @Autowired
    private PlatformAccountMapper platformAccountMapper;

    private TransactionMQConsumer() {
        try {
            this.consumer = new DefaultMQPushConsumer(CONSUMER_GROUP_NAME);
            this.consumer.setNamesrvAddr(NAME_SERVER);
            this.consumer.setConsumeThreadMin(10);
            this.consumer.setConsumeThreadMax(30);
            this.consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
            this.consumer.subscribe(TX_PAY_TOPIC, TX_PAY_TAGS);

            this.consumer.registerMessageListener(new MessageListenerConcurrentlyImpl());

            this.consumer.start();
        } catch (MQClientException e) {
            e.printStackTrace();
        }
    }

    //内部类实现消息监听器
    class MessageListenerConcurrentlyImpl implements MessageListenerConcurrently {

        @Override
        public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgList, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
            MessageExt messageExt = msgList.get(0);
            try {
                String topic = messageExt.getTopic();
                String keys = messageExt.getKeys();
                String tags = messageExt.getTags();
                //获取消息
                String msg = new String(messageExt.getBody(), RemotingHelper.DEFAULT_CHARSET);
                System.err.println("平台账户："+"topic:" + topic + "---keys:" + keys + "---tags:" + tags+"---msg:"+msg);
                Map<String, Object> params = JSONUtil.toBean(msg, Map.class);
                String userId = (String) params.get("userId");
                String orderId = (String) params.get("orderId");
                String accountId = (String) params.get("accountId");
                //采坑这么强转会报错
                //   BigDecimal money = (BigDecimal) params.get("money");
                BigDecimal money = new BigDecimal(params.get("money").toString());
                //调用平台账户mapper增加余额
                PlatformAccount pa = platformAccountMapper.selectByPrimaryKey("1");
                 pa.setCurrentBalance(pa.getCurrentBalance().add(money));
                pa.setVersion(pa.getVersion() + 1);
                Date currentTime = new Date();
                pa.setUpdateTime(currentTime);
                pa.setDateTime(currentTime);

                int updateResult = platformAccountMapper.updateByPrimaryKeySelective(pa);
                if (updateResult == 1) {
                    System.out.println("增加平台账户的余额成功");
                } else {
                    System.out.println("增加平台账户的余额失败");
                }

            } catch (UnsupportedEncodingException e) {
                e.printStackTrace();
                //msg.getReconsumeTimes();
                //	如果处理多次操作还是失败, 记录失败日志（做补偿 回顾 人工处理）
                return ConsumeConcurrentlyStatus.RECONSUME_LATER;
            }

            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        }
    }

}


